package com.roncoo.eshop.storm.bolt;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.trident.util.LRUMap;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONArray;
import com.roncoo.eshop.storm.utils.HttpClientUtils;
import com.roncoo.eshop.storm.zk.ZookeeperSession;

/**
 * Copyright: Copyright (c) 2019 Hu-Cheng
 *
 * @ClassName: ProductCountBolt.java
 * @Description: 接收解析LogParseBolt的信息，该bolt统计商品被访问的次数
 *               storm实时计算统计出热门商品数据，然后根据taskid进行分段存储到zookeeper中
 *
 * @version: v1.0.0
 * @author: hucheng
 * @date: Oct 15, 2019 1:49:23 PM
 *
 */
@SuppressWarnings("serial")
public class ProductCountBolt extends BaseRichBolt {

	private static final Logger LOGGER = LoggerFactory.getLogger(ProductCountBolt.class);

	@SuppressWarnings("unused")
	private static final long SerialVersionUID = -8761807561458126413L;

	// 使用LRUmap计算统计
	private LRUMap<Long, Long> productCountMap = new LRUMap<>(1000);

	@SuppressWarnings("unused")
	private OutputCollector collector;

	private Integer taskid;

	private ZookeeperSession zSession;

	@SuppressWarnings("rawtypes")
	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
		this.zSession = ZookeeperSession.getInstance();
		// 获取taskid，根据taskid做分段存储
		this.taskid = context.getThisTaskId();
		new Thread(new ProductCountThread()).start();
		new Thread(new HotCountThread()).start();
		initTaskId(taskid);
	}

	/**
	 * 初始化所有task的taskid，存储到zookeeper节点中去
	 * 
	 * @param taskid
	 */
	private void initTaskId(int taskid) {
		String lockPath = "/taskid-list-lock";
		zSession.acquireDistributedLockByPath(lockPath);

		String nodePath = "/taskid-list";
		zSession.createNode(nodePath);
		String taskidList = zSession.getNodeData(nodePath);
		LOGGER.info("[ProductCountBolt 获取到taskid list] taskid = " + taskidList);

		if (!taskidList.contains(taskid + "")) {
			if (!"".equals(taskidList)) {
				taskidList += "," + taskid;
			} else {
				taskidList += taskid;
			}
		}

		zSession.setNodeData(nodePath, taskidList);

		LOGGER.info("[ProductCountBolt 设置taskid list] taskidList = " + taskidList);

		zSession.releaseDistributedLock(lockPath);

	}

	/**
	 * Copyright: Copyright (c) 2019 Hu-Cheng
	 *
	 * @ClassName: ProductCountBolt.java
	 * @Description: 写一个简单的算法，保存商品访问量前三名
	 *
	 * @version: v1.0.0
	 * @author: hucheng
	 * @date: Oct 15, 2019 3:23:13 PM
	 *
	 */
	private class ProductCountThread implements Runnable {
		@Override
		public void run() {
			List<Map.Entry<Long, Long>> topnProductList = new ArrayList<Map.Entry<Long, Long>>();
			List<Long> productList = new ArrayList<Long>();
			while (true) {
				topnProductList.clear();
				productList.clear();
				int topn = 3;
				boolean bigger = false;

				for (Map.Entry<Long, Long> productCountEntry : productCountMap.entrySet()) {
					if (topnProductList.size() == 0) {
						topnProductList.add(productCountEntry);
					} else {
						for (int i = 0; i < topnProductList.size(); i++) {
							Map.Entry<Long, Long> topnProductCountEntry = topnProductList.get(i);
							if (productCountEntry.getValue() > topnProductCountEntry.getValue()) {
								topnProductList.add(i, productCountEntry);
								if (topnProductList.size() > topn) {
									topnProductList.remove(topnProductList.size() - 1);
								}
								bigger = true;
								break;
							}
						}

						if (!bigger) {
							if (topnProductList.size() < topn) {
								topnProductList.add(productCountEntry);
							}
						}
					}
				}

				for (Map.Entry<Long, Long> topnProductEntry : topnProductList) {
					productList.add(topnProductEntry.getKey());
				}

				if (productList.size() > 0) {
					// 将热门商品访问根据taskid分段的存在zookeeper中
					String topnProductListJSON = JSONArray.toJSONString(productList);
					String productListNodePath = "/task-hot-product-list-" + taskid;
					zSession.createNode(productListNodePath);
					zSession.setNodeData(productListNodePath, topnProductListJSON);
					LOGGER.info("[ProductCountBolt计算出top3热门商品列表，根据不同task分段存储] zk path = " + productListNodePath
							+ ",zk data = " + topnProductListJSON);
				}

				Utils.sleep(5000);
			}
		}
	}

	public class HotCountThread implements Runnable {
		@Override
		public void run() {
			LOGGER.info("[HotCountThread 计算热点缓存商品数据开始]");
			List<Map.Entry<Long, Long>> productList = new ArrayList<Map.Entry<Long, Long>>();
			List<Long> hotProductList = new ArrayList<Long>();
			List<Long> lastTimeHotProductIdList = new ArrayList<Long>();
			while (true) {
				try {
					// 1.将LUPmap中的数据按照访问次数进行全局排序
					productList.clear();
					hotProductList.clear();
					boolean bigger = false;

					for (Map.Entry<Long, Long> productCountEntry : productCountMap.entrySet()) {
						if (productList.size() == 0) {
							productList.add(productCountEntry);
						} else {
							for (int i = 0; i < productList.size(); i++) {
								Map.Entry<Long, Long> topnProductCountEntry = productList.get(i);
								if (productCountEntry.getValue() > topnProductCountEntry.getValue()) {
									productList.add(i, productCountEntry);
									bigger = true;
									break;
								}
							}

							if (!bigger) {
								if (productList.size() < productCountMap.size()) {
									productList.add(productCountEntry);
								}
							}
						}
					}

					LOGGER.info("[HotCountThread 待计算热点缓存商品数据] " + productList);

					if (productList.size() > 0) {
						// 2.计算出95%的商品访问次数的平均值
						int calculateCount = (int) Math.floor(productList.size() * 0.95);
						int totalCount = 0;
						for (int i = productList.size() - 1; i >= productList.size() - calculateCount; i--) {
							totalCount += productList.get(i).getValue();
						}
						// 求平均值
						int avgCount = totalCount / calculateCount;

						// 判断缓存集合里面的数据是否是平均值的10倍
						for (Map.Entry<Long, Long> productCcountEntry : productList) {
							if (productCcountEntry.getValue() > 10 * avgCount) {
								// 如果商品缓存商品集合里的数据，是95%商品的平均值的10倍，那么就认为他是热点数据
								hotProductList.add(productCcountEntry.getKey());
								// 如果热点缓存已经存在，则不执行以下操作
								if (!lastTimeHotProductIdList.contains(productCcountEntry.getKey())) {
									LOGGER.info("[ProductCountBolt计算出热点商品列表] " + hotProductList);
									// 3.将热点缓存数据反推送给nginx
									// 先将数据反推送给nginx分发层
									String distributeNginxURL = "http://192.168.124.204/hot?productId="
											+ productCcountEntry.getKey();
									HttpClientUtils.doGet(distributeNginxURL);

									// 根据热点缓存商品id，先去获取完整的商品信息，再将商品信息范推送给nginx应用层
									String cacheServiceURL = "http://192.168.124.31:8081/getProductInfo?productId="
											+ productCcountEntry.getKey();
									String response = HttpClientUtils.doGet(cacheServiceURL);

									String[] appNginxURLs = { "http://192.168.124.205", "http://192.168.124.206" };

									for (int i = 0; i < appNginxURLs.length; i++) {
										Map<String, String> param = new HashMap<String, String>();
										param.put("productId", productCcountEntry.getKey() + "");
										param.put("productInfo", response);
										HttpClientUtils.doGet(appNginxURLs[0], param);
									}
								}

							}
						}

						LOGGER.info("[HotCountThread 热点缓存数据列表] hotProductList = " + hotProductList);
						LOGGER.info(
								"[HotCountThread 老的热点缓存数据列表] lastTimeHotProductIdList = " + lastTimeHotProductIdList);
						// 4.实时感应热点缓存消失的数据
						if (lastTimeHotProductIdList.size() == 0) {
							if (hotProductList.size() > 0) {
								for (Long productId : hotProductList) {
									lastTimeHotProductIdList.add(productId);
								}
								LOGGER.info("[HotCountThread 保存上次热点数据] lastTimeHotProductIdList = "
										+ lastTimeHotProductIdList);
							}
						} else {
							for (Long productId : lastTimeHotProductIdList) {
								if (!hotProductList.contains(productId)) {
									// 当发现之前保存的热点缓存数据不再现在的热点缓存数据列表中，就表示该商品数据已经不是热点缓存了
									// 取消热点缓存商品，发送一个http请求，告诉nginx分发层，该商品不是热点缓存
									String url = "http://192.168.124.204/cancel_hot?productId=" + productId;
									HttpClientUtils.doGet(url);
									LOGGER.info("[HotCountThread 取消热点缓存数据 ]" + productId);
								}
							}

							if (hotProductList.size() > 0) {
								lastTimeHotProductIdList.clear();
								for (Long productId : hotProductList) {
									lastTimeHotProductIdList.add(productId);
								}
								LOGGER.info("[HotCountThread 保存上次热点数据] lastTimeHotProductIdList = "
										+ lastTimeHotProductIdList);
							} else {
								lastTimeHotProductIdList.clear();
							}
						}
					}

					Utils.sleep(5000);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}

	}

	@Override
	public void execute(Tuple input) {
		Long productId = input.getLongByField("productId");

		Long count = productCountMap.get(productId);
		if (count == null) {
			count = 0L;
		}
		count++;
		productCountMap.put(productId, count);
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		
	}

}
